package streaming

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}

object srteaming_wordcount {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("_01StreamingWordCountApp")
      .setMaster("local[*]")
    if(args == null || args.length != 2) {
      println(
        """
          |Usage: <host> <port>
        """.stripMargin)
      System.exit(-1)
    }
      val Array(host, port) = args
    val bi=Seconds(2)
    val ssc=new StreamingContext(conf,bi)
      //主机端口号    设置副本为两个
   val input=ssc.socketTextStream(host,port.toInt,storageLevel = StorageLevel.MEMORY_AND_DISK_SER_2)
    val ret=input.flatMap(line=>line.split("\\s+"))
      .map(work=>(work,1)).reduceByKey(_+_).print()
    ssc.start()
    ssc.awaitTermination()
  }
}

